package com.ct.ecommerce.idgen.service;

import com.ct.ecommerce.api.idgen.client.SnGeneratorService;
import com.ct.ecommerce.framework.enums.SerialNumberTypeEnumEnum;
import com.ct.ecommerce.idgen.core.Segment;
import com.ct.ecommerce.idgen.core.SegmentBuffer;
import com.ct.ecommerce.idgen.dao.entity.SnStrategyEntity;
import com.ct.ecommerce.idgen.dao.mapper.SnStrategyMapper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.StringJoiner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @author changtong
 * @since 2021/5/24
 */
@DubboService
@Slf4j
public class SnGeneratorServiceImpl implements SnGeneratorService {

    /**
     * 最大步长不超过100,0000
     */
    private static final int MAX_STEP = 1000000;
    /**
     * 一个Segment维持时间为15分钟
     */
    private static final long SEGMENT_DURATION = 15 * 60 * 1000L;

    private volatile boolean initOk = false;

    @Autowired
    private Executor taskExecutor;

    private Map<String, SegmentBuffer> cache = new ConcurrentHashMap<>();

    @Resource
    private SnStrategyMapper snStrategyMapper;

    @PostConstruct
    public void init() {
        log.info("SnGeneratorServiceImpl Init ...");
        //初始化SegmentBuffer
        updateCacheFromDb();
        initOk = true;
        scheduledUpdateCacheFromDb();
    }

    private void scheduledUpdateCacheFromDb() {

        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
            .setDaemon(true)
            .setNameFormat("check-snCache-thread").build();

        ScheduledThreadPoolExecutor scheduledPool
            = new ScheduledThreadPoolExecutor(1, namedThreadFactory,
            new ThreadPoolExecutor.AbortPolicy());

        //每5分钟更新下cache
        scheduledPool.scheduleWithFixedDelay(this::updateCacheFromDb,
            5, 5,
            TimeUnit.MINUTES);

    }

    private void updateCacheFromDb() {
        try {
            Optional.ofNullable(snStrategyMapper.selectAll())
                .filter(dbTags -> !dbTags.isEmpty())
                .ifPresent(dbTags -> {
                    dbTags.stream()
                        .filter(dbTag -> !cache.containsKey(dbTag))
                        .map(dbTag -> {
                            SegmentBuffer buffer = new SegmentBuffer();
                            buffer.setKey(dbTag);
                            Segment segment = buffer.getCurrent();
                            segment.setValue(new AtomicLong(0));
                            segment.setMax(0);
                            segment.setStep(0);
                            return buffer;
                        }).forEachOrdered(buf -> cache.put(buf.getKey(), buf));

                    List<String> cacheTags = new ArrayList<>(cache.keySet());
                    Set<String> dbTagSet = new HashSet<>(dbTags);

                    cacheTags.stream().filter(t -> !dbTagSet.contains(t))
                        .forEach(t -> {
                            cache.remove(t);
                            log.info("Remove tag {} from IdCache", t);
                        });
                });
        } catch (Exception e) {
            log.error("updateCacheFromDb error : ", e);
        }
    }


    @Override
    public String get(Integer serialNumberType) {
        String bizKey = SerialNumberTypeEnumEnum.valueOf(serialNumberType).getShowName();
        if (!initOk || !cache.containsKey(bizKey)) {
            throw new RuntimeException("获取业务key【" + bizKey + "】sn失败....");
        }
        SegmentBuffer buffer = cache.get(bizKey);
        if (!buffer.isInitOk()) {
            synchronized (buffer) {
                if (!buffer.isInitOk()) {
                    try {
                        initSegmentFromDb(bizKey, buffer.getCurrent());
                        log.info("Init buffer. Update bizKey {} {} from db", bizKey,
                            buffer.getCurrent());
                        buffer.setInitOk(true);
                    } catch (Exception e) {
                        log.error("Init buffer {} exception", buffer.getCurrent(), e);
                    }
                }
            }
        }
        return getIdFromSegmentBuffer(bizKey);
    }

    private String getIdFromSegmentBuffer(String bizKey) {
        SegmentBuffer buffer = cache.get(bizKey);
        for (; ; ) {
            log.info("buffer = {}", buffer);
            buffer.rLock().lock();
            try {
                final Segment segment = buffer.getCurrent();
                if (!buffer.isNextReady() && (segment.getIdle() < 0.9 * segment.getStep()) && buffer
                    .getThreadRunning().compareAndSet(false, true)) {
                    taskExecutor.execute(() -> {
                        Segment next = buffer.getSegments()[buffer.nextPos()];
                        boolean updateOk = false;
                        try {
                            initSegmentFromDb(buffer.getKey(), next);
                            updateOk = true;
                            log.info("update segment {} from db {}", buffer.getKey(), next);
                        } catch (Exception e) {
                            log.error(buffer.getKey() + " updateSegmentFromDb exception", e);
                        } finally {
                            if (updateOk) {
                                buffer.wLock().lock();
                                buffer.setNextReady(true);
                                buffer.getThreadRunning().set(false);
                                buffer.wLock().unlock();
                            } else {
                                buffer.getThreadRunning().set(false);
                            }
                        }
                    });
                }
                long value = segment.getValue().getAndIncrement();
                if (value < segment.getMax()) {
                    return getSn(bizKey, value);
                }
            } finally {
                buffer.rLock().unlock();
            }
            waitAndSleep(buffer);
            buffer.wLock().lock();
            try {
                final Segment segment = buffer.getCurrent();
                long value = segment.getValue().getAndIncrement();
                if (value < segment.getMax()) {
                    return getSn(bizKey, value);
                }
                if (buffer.isNextReady()) {
                    buffer.switchPos();
                    buffer.setNextReady(false);
                } else {
                    log.error("Both two segments in {} are not ready!", buffer);
                    throw new RuntimeException("获取业务key【" + bizKey + "】sn失败....");
                }
            } finally {
                buffer.wLock().unlock();
            }
        }
    }

    private void waitAndSleep(SegmentBuffer buffer) {
        int roll = 0;
        while (buffer.getThreadRunning().get()) {
            roll += 1;
            if (roll > 10000) {
                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                    break;
                } catch (InterruptedException e) {
                    log.info("Thread {} Interrupted", Thread.currentThread().getName());
                    break;
                }
            }
        }
    }

    private String getSn(String bizKey, long value) {
        Calendar cal = Calendar.getInstance();
        String year = String.format("%s", cal.get(Calendar.YEAR));
        String month = String.format("%02d", cal.get(Calendar.MONTH) + 1);
        String date = String.format("%02d", cal.get(Calendar.DATE));

        StringJoiner sn = new StringJoiner("");

        sn.add(bizKey).add(year).add(month)
            .add(date);

        if (value < 10000L){
            sn.add(String.format("%04d",value));
        }else {
            sn.add(String.valueOf(value));
        }

        return sn.toString();

    }

    private void initSegmentFromDb(String bizKey, Segment segment) {
        SegmentBuffer buffer = segment.getBuffer();
        SnStrategyEntity snStrategy;
        if (!buffer.isInitOk()) {
            //没有初始化过
            snStrategy = this.updateMaxIdAndGetLeafAlloc(bizKey);
            buffer.setStep(snStrategy.getStep());
            buffer.setMinStep(snStrategy.getStep());
        } else if (buffer.getUpdateTimestamp() == 0) {
            snStrategy = this.updateMaxIdAndGetLeafAlloc(bizKey);
            buffer.setUpdateTimestamp(System.currentTimeMillis());
            buffer.setStep(snStrategy.getStep());
            buffer.setMinStep(snStrategy.getStep());
        } else {
            long duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();
            int nextStep = buffer.getStep();
            if (duration < SEGMENT_DURATION) {
                if (nextStep * 2 > MAX_STEP) {
                    //do nothing
                } else {
                    //短时间内 提升步长翻一倍
                    nextStep = nextStep * 2;
                }
            } else if (duration < SEGMENT_DURATION * 2) {
                //do nothing with nextStep
            } else {
                //稳定后再更改回来
                nextStep = nextStep / 2 >= buffer.getMinStep() ? nextStep / 2 : nextStep;
            }
            log.info("bizKey[{}], step[{}], duration[{}mins], nextStep[{}]",
                bizKey, buffer.getStep(), String.format("%.2f", ((double) duration / (1000 * 60))),
                nextStep);

            snStrategy = this.updateMaxIdByCustomStepAndGetLeafAlloc(bizKey, nextStep);
            buffer.setUpdateTimestamp(System.currentTimeMillis());
            buffer.setStep(nextStep);
            buffer.setMinStep(snStrategy.getStep());
        }
        // must set value before set max
        long value = snStrategy.getMaxId() - buffer.getStep();
        segment.getValue().set(value);
        segment.setMax(snStrategy.getMaxId());
        segment.setStep(buffer.getStep());

    }

    private SnStrategyEntity updateMaxIdAndGetLeafAlloc(String bizKey) {
        snStrategyMapper.updateMaxId(bizKey);
        return snStrategyMapper.selectById(bizKey);
    }

    private SnStrategyEntity updateMaxIdByCustomStepAndGetLeafAlloc(String bizKey, int step) {
        snStrategyMapper.updateMaxIdByCustomStepAndGetLeafAlloc(bizKey, step);
        return snStrategyMapper.selectById(bizKey);
    }


}
